package cn.yj.netty.support.test;

import cn.yj.netty.message.convert.IProtocolMessageConvert;
import cn.yj.netty.message.heart.IHeartBeatMonitor;
import cn.yj.netty.message.outIn.SimpleChannelOutboundHandler;
import cn.yj.netty.server.AbstractNettyServer;
import cn.yj.netty.common.Const;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.ObjectUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * @author 永健
 * @since 2022-01-17 16:53
 */
public class NettyServer extends AbstractNettyServer {


    private static final Logger log = LoggerFactory.getLogger(NettyServer.class);

    private static InetSocketAddress socketAddress;

    public static void setSocketAddress(InetSocketAddress socketAddress) {
        NettyServer.socketAddress = socketAddress;
    }

    @Autowired
    private IProtocolMessageConvert protocolMessageConvert;

    @Resource
    private IHeartBeatMonitor heartBeatMonitor;

    @PostConstruct
    public void init() {
        super.start(socketAddress);
    }

    @Override
    protected void run(InetSocketAddress inetSocketAddress) {
        EventLoopGroup parentGroup = new NioEventLoopGroup(1);
        EventLoopGroup childGroup = new NioEventLoopGroup(100);
        try {
            ServerBootstrap serverBootstrap =
                    new ServerBootstrap()
                            .group(parentGroup, childGroup)
                            .channel(NioServerSocketChannel.class)
                            .childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {

                    // 粘包处理
                    // 解码方式不是按照 [] 前后包围的方式解码，是按照每个分隔符来解码的
                    ByteBuf end = Unpooled.copiedBuffer(Const.DELIMITER_END.getBytes());
                    socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, false, end));

                    // 解码处理器
                    socketChannel.pipeline().addLast("decoder", new MessageDecode());
                    // 编码处理器
                    socketChannel.pipeline().addLast("encoder", new MessageEncode());

                    // 心跳检测规则
                    socketChannel.pipeline().addLast(new IdleStateHandler(Const.INTERVAL_TIME, 0, 0, TimeUnit.SECONDS));


                    // 消息处理器-读取
                    socketChannel.pipeline().addLast(new SimpleChannelInboundHandler<String>() {
                        @Override
                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                            super.exceptionCaught(ctx, cause);
                            log.info("netty => 异常,host={}",ctx.channel().remoteAddress().toString());
                        }

                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            super.channelActive(ctx);
                            log.info("netty => 客户端连接,host={}",ctx.channel().remoteAddress().toString());
                        }

                        @Override
                        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
                            ctx.channel().close();
                            super.handlerRemoved(ctx);
                        }

                        @Override
                        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
                            log.info("===> 接收数据，协议包原内容,host={},内容 ==>\n{}", ctx.channel().remoteAddress().toString(),msg);
                            if (ObjectUtils.isEmpty(msg)){
                                log.info("协议包原内容为空,拒绝处理,关闭连接,host={},内容 = {}", ctx.channel().remoteAddress().toString(),msg);
                                ctx.channel().close();
                            }
                            protocolMessageConvert.message(ctx, msg);
                        }

                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                            heartBeatMonitor.userEventTriggered(ctx,evt,()-> {
                                try {
                                    super.userEventTriggered(ctx, evt);
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            });

                        }
                    });

                    // 消息写处理
                    socketChannel.pipeline().addLast(new SimpleChannelOutboundHandler());
                }

            }).localAddress(inetSocketAddress).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true).childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(64 * 1024, 128 * 1024));

            ChannelFuture f = serverBootstrap.bind(inetSocketAddress.getPort()).sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            parentGroup.shutdownGracefully();
            childGroup.shutdownGracefully();
        }
    }

}
